package t20240425;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Test1 {


    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);


        String sourceSql = "CREATE TABLE datagen (" +
                "    stime INT," +
                "    sn STRING," +
                "    ts AS localtimestamp" +
                "  ) WITH (" +
                "    'connector' = 'datagen'," +
                "    'rows-per-second'='1'," +
                "    'fields.stime.kind'='sequence'," +
                "    'fields.stime.start'='1'," +
                "    'fields.stime.end'='10000'," +
                "    'fields.sn.length'='10'" +
                "  )";

        String sinkSql =
                "  CREATE TABLE print_table (" +
                "    stime INT," +
                "    sn STRING," +
                "    stime_ext STRING" +
                "    ) WITH (" +
                "    'connector' = 'print'" +
                "  )";

        String execSql = "INSERT INTO print_table select stime,sn" +
                ",DATE_FORMAT(ts, 'yyyy-MM-dd hh:mm:ss') as stime_ext" +
//                ",convert_tz(DATE_FORMAT(ts, 'yyyy-MM-dd hh:mm:ss'), 'UTC', 'Asia/Shanghai') as stime_ext" +
//                " from datagen;
                " from datagen " +
                "where (SUBSTRING(sn, -1, 1) < '0' OR SUBSTRING(sn, -1, 1) > '9')";

        tableEnvironment.executeSql(sourceSql);
        System.out.println("-----1");
        tableEnvironment.executeSql(sinkSql);
        System.out.println("-----2");
        tableEnvironment.executeSql(execSql);


    }

}
